package com.tech.customer

import com.tech.common.KuduUtil
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession

import scala.collection.mutable

class LoadResourceManager extends Serializable {

  var ruleMapBroadcast: Broadcast[mutable.HashMap[String, mutable.HashMap[String, String]]] = _
  var retaileventSchemaMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _
  var customerSchemaMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _

  def unPersist(): Unit = {
    ruleMapBroadcast.unpersist(true)
  }

  def load(spark: SparkSession) {
    val ruleMap = KuduUtil.getRule()
    ruleMapBroadcast = spark.sparkContext.broadcast(ruleMap)
  }

  def loadOnce(spark: SparkSession) {
    val retaileventSchemaMap = KuduUtil.getTableSchema("retailevent")
    val customerSchemaMap = KuduUtil.getTableSchema("customer")

    retaileventSchemaMapBroadcast = spark.sparkContext.broadcast(retaileventSchemaMap)
    customerSchemaMapBroadcast = spark.sparkContext.broadcast(customerSchemaMap)
  }
}
